package com.ask.mq;

import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;

@Component
public class RedisStreamUtil {
    private final StringRedisTemplate stringRedisTemplate;

    // Bean实例化时执行构造函数(StringRedisTemplate已经存在与Spring容器中)
    public RedisStreamUtil(StringRedisTemplate stringRedisTemplate) {
        this.stringRedisTemplate = stringRedisTemplate;
    }

    /**
     * 读取 Redis Stream 消息
     * @param queueName 消息队列名称
     * @param consumerGroup 消费者组名称
     * @param consumerName 消费者名称
     * @param count 读取的消息数量
     * @param blockTimeout 阻塞时间
     * @return 消息列表
     */
    public List<MapRecord<String, Object, Object>> readStreamMessages(
            String queueName,
            String consumerGroup,
            String consumerName,
            long count,
            Duration blockTimeout
    ){
        return stringRedisTemplate.opsForStream().read(
                Consumer.from(consumerGroup, consumerName), // 指定消费之组消费者
                StreamReadOptions.empty().count(count).block(blockTimeout),
                StreamOffset.create(queueName, ReadOffset.lastConsumed()) // 从上次消费的位置读取
        );
    }
}
